package com.behavioranalysis.flinkprogram.flink.ad;

import com.behavioranalysis.flinkprogram.conf.ConfigurationManager;
import com.behavioranalysis.flinkprogram.constant.Constants;
import com.behavioranalysis.flinkprogram.dao.IAdBlacklistDAO;
import com.behavioranalysis.flinkprogram.dao.IAdUserClickCountDAO;
import com.behavioranalysis.flinkprogram.dao.factory.DAOFactory;
import com.behavioranalysis.flinkprogram.dao.output.AdClickTrendSink;
import com.behavioranalysis.flinkprogram.dao.output.AdRealTimeStatSink;
import com.behavioranalysis.flinkprogram.dao.output.BlackSink;
import com.behavioranalysis.flinkprogram.dao.output.DailyUserAdClickCountSink;
import com.behavioranalysis.flinkprogram.domain.AdBlacklist;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 广告点击流量实时统计Flink作业
 */
public class AdClickRealTimeStatFlink {
    /*public static void main(String[] args) throws Exception {

        // 构建flink流式处理程序的上下文，流式的话使用StreamExecutionEnvironment，批处理使用ExecutionEnvironment
        // 本地测试使用local environment（本地执行环境）
        StreamExecutionEnvironment env = null;
        if(ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            env = StreamExecutionEnvironment.createLocalEnvironment();
        } else {
            env = StreamExecutionEnvironment.getExecutionEnvironment();
        }
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 使用socket源，流处理程序处理的是DataStream数据集，批处理程序处理的是DataSet数据集
        DataStream<String> adRealTimeLog = env
                .socketTextStream("192.168.0.111", 9999);

        DataStream<Tuple6<Long, String, String, String, String, String>> adRealTimeLogTuple = adRealTimeLog.map(
                new MapFunction<String, Tuple6<Long, String, String, String, String, String>>() {
                    private SimpleDateFormat simpleDateFormat =
                            new SimpleDateFormat("yyyyMMdd");
                    @Override
                    public Tuple6<Long, String, String, String, String, String> map(String log) throws Exception {
                        String[] splitedLogs = log.split(" ");
                        long timepstamp = Long.valueOf(splitedLogs[0]);
                        String province = splitedLogs[1];
                        String city = splitedLogs[2];
                        String userid = splitedLogs[3];
                        String adid = splitedLogs[4];
                        // 提取日期(yyyyMMdd)、userid、adid
                        Date date = new Date(timepstamp);
                        String datekey = simpleDateFormat.format(date);

                        // 拼接key
                        String key = datekey + "_" + userid + "_" + adid;

                        return new Tuple6<>(timepstamp, province, city, userid, adid, key);
                    }
                }
        );

        //adRealTimeLogTuple.print();

        // 侧面输出流： clickCount >= 100 的用户userid
        final OutputTag<Long> blacklistUserid = new OutputTag<Long>("blacklistUserid"){};


        *//**
     * 一： 设置滚动窗口，进行黑名单用户过滤
     *      窗口大小：两分钟，120秒
     *      数据库黑名单表按流来读，行不通
     *      按静态数据来读
     *//*

        DataStream<Tuple6<Long,String,String,String,String,String>> watermarkStream = adRealTimeLogTuple
                .assignTimestampsAndWatermarks(
                        // 1.分配时间戳，并生成水印
                        new AssignerWithPeriodicWatermarks<Tuple6<Long, String, String, String, String, String>>() {
                            private long currentMaxTimestamp = 0L;
                            private long maxOutOfOrderness = 5000L;//最大乱序时间为10s

                            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

                            @Nullable
                            @Override
                            public Watermark getCurrentWatermark() {
                                return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
                            }

                            @Override
                            public long extractTimestamp(Tuple6<Long, String, String, String, String, String> element, long previousElementTimestamp) {
                                long timestamp = element.f0;
                                currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);

                                *//**
     *
     *//*
                                System.out.println("timestamp:"
                                        + element.f0 + "|" + format.format(element.f0) + ","
                                        + currentMaxTimestamp + "|" + format.format(currentMaxTimestamp) + ","
                                        + getCurrentWatermark().toString());
                                return timestamp;
                            }
                        }
                );

        DataStream<Tuple6<Long,String,String,String,String,String>> filteredAdRealTimeLogStream = watermarkStream
                // 这里弃用了keyby，没必要，还浪费时间，每个key都要查一次数据库。
                .timeWindowAll(Time.seconds(3))
                .process(new ProcessAllWindowFunction<Tuple6<Long,String,String,String,String,String>, Tuple6<Long,String,String,String,String,String>, TimeWindow>() {
                    private static final long serialVersionUID = 1L;
                    @Override
                    public void process(Context context, Iterable<Tuple6<Long, String, String, String, String, String>> elements, Collector<Tuple6<Long, String, String, String, String, String>> out) throws Exception {
                        // 将blacklist全部查出来
                        IAdBlacklistDAO adBlacklistDAO = DAOFactory.getAdBlackListDAO();
                        List<AdBlacklist> adBlacklists = adBlacklistDAO.findAll();

                        List<String> blacklist = new ArrayList<>();
                        for(AdBlacklist adBlacklist : adBlacklists) {
                            blacklist.add(String.valueOf(adBlacklist.getUserid()));
                        }

                        //System.out.println("黑名单总计:" + blacklist.size());

                        // 判断此窗口中的记录的userid是否在blacklist中，不在就output
                        for(Tuple6<Long, String, String, String, String, String> tuple6 : elements) {
                            String userid = tuple6.f3;
                            if(blacklist.contains(userid) == false) {
                                // 2. 将此记录collect
                                out.collect(tuple6);
                            }
                        }
                    }
                });
                *//**
     *  统计每天每个用户对每个广告的点击量
     *//*

       SingleOutputStreamOperator<Tuple4<Long, Long, String, Long>> dailyUserAdClickCount = filteredAdRealTimeLogStream
                .keyBy(new KeySelector<Tuple6<Long,String,String,String,String,String>, String>() {
                    @Override
                    public String getKey(Tuple6<Long, String, String, String, String, String> value) throws Exception {
                        return value.f5;
                    }
                })
                .timeWindow(Time.seconds(3))
                .process(new ProcessWindowFunction<Tuple6<Long,String,String,String,String,String>, Tuple4<Long, Long, String, Long>, String, TimeWindow>() {
                    private static final long serialVersionUID = 1L;
                    @Override
                    public void process(String key, Context context, Iterable<Tuple6<Long, String, String, String, String, String>> elements, Collector<Tuple4<Long, Long, String, Long>> out) throws Exception {
                        // 相同的key 的tuple会进到这里面，
                        // 那么我们就可以将此窗口中此key的count统计出来存入数据库 <yyyyMMdd_userid_adid, count>
                        // 都是一个Key,那么要出如数据库的就是Key,elements的大小
                        // 在这里存数据的话，也是每个key需连接一次数据库，
                        //System.out.println("此key是：" + key);

                        // 2. 这里还需要执行第二个，查询这一天，这一用户，对这一广告的点击量是否已经超过100
                        // 这里刚好是按key分组的，我们就是要用key进行查询
                        // date, userid, adid
                        // count >= 100 将userid返回 拉入黑名单
                        // 这个主流我们要用来addSink到数据库
                        // 所以将黑名单用户做成侧面输出流，然后在外面进行去重，保存到数据库。
                        String[] keySplited = key.split("_");
                        // yyyyMMdd -> yyyy-MM-dd
                        SimpleDateFormat format =
                                new SimpleDateFormat("yyyyMMdd");
                        SimpleDateFormat dateFormat =
                                new SimpleDateFormat("yyyy-MM-dd");
                        String date = dateFormat.format(format.parse(keySplited[0]));
                        long userid = Long.valueOf(keySplited[1]);
                        long adid = Long.valueOf(keySplited[2]);

                        IAdUserClickCountDAO adUserClickCountDAO = DAOFactory.getAdUserClickCountDAO();
                        int clickCount = adUserClickCountDAO.findClickCountByMultiKey(
                                date, userid, adid);

                        // 到这里就是将每一个窗口中，相同的key聚合成了key counts
                        // 1. 返回的是<userid, adid, date,counts>，那我们可以直接将此流保存到数据库
                        // 需要插入的插入，需要更新的更新
                        // TODO
                        out.collect(new Tuple4<>(userid, adid,date,Long.valueOf(IteratorUtils.toList(elements.iterator()).size())));

                        // 判断，如果点击量大于等于100，就拉入黑名单
                        if(clickCount >= 100) {
                            // 将此userid加入黑名单侧面输出流
                            context.output(blacklistUserid,
                                    userid);
                        }
                    }
                });
        *//**
     * 将dailyUserAdClickCount 流 addSink到数据库
     * 数据库端 要进行 插入？还是更新？的判别
     *//*
        dailyUserAdClickCount.print();
        dailyUserAdClickCount.addSink(new DailyUserAdClickCountSink());

        *//**
     * 得到 blacklistUserid 侧面输出流，并将其userid去重，然后保存到数据库
     * 继续windowALl
     *//*
        DataStream<Long> blacklistUseridStream = dailyUserAdClickCount.getSideOutput(blacklistUserid);

        blacklistUseridStream
                .timeWindowAll(Time.seconds(3))
                .process(new ProcessAllWindowFunction<Long, Long, TimeWindow>() {
                    private static final long serialVersionUID = 1L;
                    @Override
                    public void process(Context context, Iterable<Long> elements, Collector<Long> out) throws Exception {
                        // 这个方法会进来此窗口中全部的black list userid,然后进行去重,最后addSink
                        // 没问题！就在这里去重，然后addSink到Mysql
                        Set<Long> distinctUserid = new HashSet<>();
                        distinctUserid.addAll(IteratorUtils.toList(elements.iterator()));
                        for (Long blackUserid : distinctUserid) {
                            out.collect(blackUserid);
                        }
                    }
                })
                .addSink(new BlackSink());
        blacklistUseridStream.print();

        *//**
     * 得到 filteredAdRealTimeLog 侧面输出流，以供后面的业务使用 1461756862000
     *
     * 注意： 这个侧面输出流在这得不到是因为
     *      它是从第一个window的process方法中返回的，
     *      我们却从最后一个window返回的SingleOutputStreamOperator中得到它
     *      显然是不可能的
     *      所以第一种方法是：我们要把第一个window产生的流单独得到，才能从中得到它
     *      第二种： 我们也可以从第二个窗口中返回它，第二个窗口中也是经过过滤的正常点击
     *      第三种： 也可以将两个窗口隔开，第一个窗口得到的就是经过过滤的正常点击filteredAdRealTimeLogStream
     *              这个filteredAdRealTimeLogStream后面的业务可以用到
     *              第二个窗口可以用filteredAdRealTimeLogStream开启
     *              从而得到每天每个用户对每个广告的点击量dailyUserAdClickCount
     *              再从dailyUserAdClickCount中将侧面输出流blacklistUseridStream返回
     *
     *      前面两种方法都是又将filteredAdRealTimeLogStream作为侧面输出流返回一次
     *      显然多此一举。
     *
     *//*

     *//**
     * 业务功能一：计算广告流量实时统计结果(yyyyMMdd_province_city_adid, clickCount)
     *  最粗
     *//*

        // 状态dailyAdClickCount侧面输出流
        final OutputTag<Tuple2<String, Long>> dailyAdClickCountGlobal = new OutputTag<Tuple2<String, Long>>("dailyAdClickCountGlobal"){};

        SingleOutputStreamOperator<Tuple5<String, String, String, Long, Long>> adRealTimeStatStream = filteredAdRealTimeLogStream
                .map(new MapFunction<Tuple6<Long, String, String, String, String, String>, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(Tuple6<Long, String, String, String, String, String> value) throws Exception {
                        SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd");
                        Date date = new Date(value.f0);
                        String datekey = format.format(date);
                        String province = value.f1;
                        String city = value.f2;
                        long adid = Long.valueOf(value.f4);
                        String key = datekey + "_" + province + "_" + city + "_" + adid;
                        return new Tuple2<>(key, 1L);
                    }
                })
                .keyBy(new KeySelector<Tuple2<String,Long>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Long> value) throws Exception {
                        return value.f0;
                    }
                })
                .timeWindow(Time.seconds(3))
                .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple5<String, String, String, Long, Long>, String, TimeWindow>() {
                    // 状态
                    // TODO 要定期清理 状态
                    private transient MapState<String, Long> dailyAdClickCount;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        MapStateDescriptor<String, Long> descriptor =
                                new MapStateDescriptor<String, Long>(
                                        "dailyAdClickCount",
                                        String.class, Long.class);
                        dailyAdClickCount = getRuntimeContext().getMapState(descriptor);
                    }

                    @Override
                    public void process(String key, Context context, Iterable<Tuple2<String, Long>> elements, Collector<Tuple5<String, String, String, Long, Long>> out) throws Exception {
                        // 1. 先将key value更新到状态中
                        //累加count
                        long count = 0L;
                        for (Tuple2<String, Long> tuple2 : elements) {
                            count += tuple2.f1;
                        }
                        // 判断此key是否已经存在于状态dailyAdClickCount中
                        if (dailyAdClickCount.contains(key)) {
                            // 存在，则累加count
                            dailyAdClickCount.put(key, dailyAdClickCount.get(key) + count);
                        } else {
                            // 不存在，则put
                            dailyAdClickCount.put(key, count);
                        }
                        // 2. 将此key在状态中的value得到并且返回
                        String[] keySplited = key.split("_");
                        String date = keySplited[0];
                        String province = keySplited[1];
                        String city = keySplited[2];
                        long adid = Long.valueOf(keySplited[3]);
                        out.collect(new Tuple5<>(date, province, city, adid, dailyAdClickCount.get(key)));

                        *//**
     * 每天每个省份每个城市对某个广告的点击量
     * 集群中的内存中保存了一份（状态）：集群中的我们要找到实时的每个省份的top3广告
     * mysql数据库中也保存了一份：数据库中的我们要在前端展示
     *
     * 将全局的状态设置为 侧面输出流
     *
     * 这个方法是 一个窗口中的每个key 都要执行一次
     * 那说明，每个key里面都要将全局的状态得到
     * 然后再去计算每个省份的top3，有点频繁。
     * TODO 能不能每个窗口计算一次，而不是每个key
     *
     * 并且计算实时的每个省份的top3广告，然后保存到数据库
     *//*

//                        for (Map.Entry<String, Long> entry : dailyAdClickCount.entries()) {
//                            context.output(dailyAdClickCountGlobal, new Tuple2<>(entry.getKey(), entry.getValue()));
//                        }

                    }
                });
        // 将窗口中的key的最新的全局count addSink到数据库，有则覆盖，无则插入
        adRealTimeStatStream.addSink(new AdRealTimeStatSink());
        adRealTimeStatStream.print();


        *//**
     * 业务功能二：实时统计每天每个省份top3热门广告
     *  统计的稍微细一些了
     *//*
     *//*DataStream<Tuple2<String, Long>> dailyAdClickCountGlobalStream = adRealTimeStatStream.getSideOutput(dailyAdClickCountGlobal);
        dailyAdClickCountGlobalStream.print();*//*

     *//**
     * 业务功能三：实时统计每天每个广告在最近1小时的滑动窗口内的点击趋势（每分钟的点击量）
     * 统计的非常细了
     * 我们每次都可以看到每个广告，最近一个小时内，每分钟内的点击量
     * 每支广告的点击趋势
     * watermarkStream ->  AdClickTrendSink
     * <date, hour, minute, adid, clickCount>
     *//*
        watermarkStream
                .map(new MapFunction<Tuple6<Long,String,String,String,String,String>, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(Tuple6<Long, String, String, String, String, String> value) throws Exception {
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmm");
                        String timeMinute = sdf.format(new Date(value.f0));
                        long adid = Long.valueOf(value.f4);
                        
                        return new Tuple2<>(timeMinute + "_" + adid, 1L);
                    }
                })
                .keyBy(new KeySelector<Tuple2<String,Long>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Long> value) throws Exception {
                        return value.f0;
                    }
                })
                .timeWindow(Time.seconds(15), Time.seconds(5))
                .sum(1)
                .map(new MapFunction<Tuple2<String,Long>, Tuple5<String, String, String, Long, Long>>() {
                    @Override
                    public Tuple5<String, String, String, Long, Long> map(Tuple2<String, Long> value) throws Exception {
                        SimpleDateFormat dateFormat =
                                new SimpleDateFormat("yyyyMMdd");
                        SimpleDateFormat dateFormat1 =
                                new SimpleDateFormat("yyyy-MM-dd");

                        String[] keySplited = value.f0.split("_");
                        long adid = Long.valueOf(keySplited[1]);
                        long clickCount = value.f1;
                        String dateMinute = keySplited[0];

                        String date = dateFormat1.format(dateFormat.parse(
                                dateMinute.substring(0, 8)));
                        String hour = dateMinute.substring(8, 10);
                        String minute = dateMinute.substring(10);

                        return new Tuple5<>(date, hour, minute, adid, clickCount);
                    }
                })//.print();
                .addSink(new AdClickTrendSink());

        env.execute("ad");
    }*/
}
